package com.lingjtx.im.core.common;

import com.lingjtx.common.core.RedisOptions;
import com.lingjtx.common.core.token.UserDetail;
import com.lingjtx.im.core.protocol.MessageProto;
import io.netty.channel.Channel;

import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class ChannelPool {

    // 支持一人多端（Map<userId, Set<Channel>>）
    private static final Map<Long, Set<Channel>> USER_CHANNELS = new ConcurrentHashMap<>();

    // 添加连接
    public static void bind(UserDetail userDetail, Channel channel) {
        USER_CHANNELS.computeIfAbsent(userDetail.getUserId(), k -> ConcurrentHashMap.newKeySet()).add(channel);
        channel.attr(ChannelAttrKeys.USER_ID).set(userDetail.getUserId());
        channel.attr(ChannelAttrKeys.DEVICE_UID).set(userDetail.getDeviceUid());
    }

    // 解绑连接
    public static void unbind(Channel channel) {
        Long userId = getUserId(channel);
        if (userId != null) {
            Set<Channel> channels = USER_CHANNELS.get(userId);
            if (channels != null) {
                channels.remove(channel);
                if (channels.isEmpty()) {
                    USER_CHANNELS.remove(userId);
                }
            }
        }
    }

    // 获取用户ID
    public static Long getUserId(Channel channel) {
        return channel.attr(ChannelAttrKeys.USER_ID).get();
    }

    public static String getDeviceUid(Channel channel) {
        return channel.attr(ChannelAttrKeys.DEVICE_UID).get();
    }

    // 获取某用户所有连接
    public static Set<Channel> getChannels(Long userId) {
        if (userId == null) {
            return Collections.emptySet();
        }
        return USER_CHANNELS.getOrDefault(userId, Collections.emptySet());
    }

    // 点对点发送
    public static void sendToUser(Long userId, MessageProto.IMMessage message) {
        Set<Channel> channels = getChannels(userId);
        channels.forEach(ch -> {
            if (ch.isActive()) ch.writeAndFlush(message);
        });
    }

    /**
     * 群聊发送，此方法只能在单节点服务上使用，分布式环境无效，因为群成员分散在不同服务节点上
     * 群发消息使用 MQ
     */
    public static void sendToGroup(Collection<Long> members, MessageProto.IMMessage message, RedisOptions redis) {
        for (Long userId : members) {
            if (!userId.equals(message.getSenderId())
                    && SessionManager.isUserOnline(userId, redis)) { // 排除自己且该用户在线
                ChannelPool.sendToUser(userId, message);
            }
        }
    }


    // 广播
    public static void broadcast(Object msg) {
        USER_CHANNELS.values().forEach(set -> set.forEach(ch -> {
            if (ch.isActive()) ch.writeAndFlush(msg);
        }));
    }
}
